package com.luojing.springcloud.config.alimq;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * Created by luojing on 2017/7/4.
 */
@Configuration
public class ProductConfig {

    @Bean
    public Producer1 initProducer(){
        if (System.getProperty("java.security.auth.login.config") == null) {
            System.setProperty("java.security.auth.login.config",  ProductConfig.class.getClassLoader().getResource("kafka_client_jaas.conf").getPath());
        }

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "SASL_SSL://kafka-ons-internet.aliyun.com:8080");
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, ProductConfig.class.getClassLoader().getResource("kafka.client.truststore.jks").getPath());
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        props.put(SaslConfigs.SASL_MECHANISM, "ONS");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);

        Producer1 producer1 = new Producer1();
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        producer1.setProducer(producer);
        return producer1;
    }

}
